package raft.service.impl;


import lombok.Getter;
import lombok.Setter;
import org.rocksdb.RocksDBException;
import raft.ThreadUtil.RaftThreadPool;
import raft.entity.*;
import raft.exception.RaftRemotingException;
import raft.node.NodeConfig;
import raft.node.NodeStatus;
import raft.node.PeerSet;
import raft.rpc.*;
import raft.service.Consensus;
import raft.service.LogModule;
import raft.service.Node;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

/**
 * 选举者必须不是 leader。
 * 必须超时了才能选举，具体超时时间根据你的设计而定,注意，每个节点的超时时间不能相同，应当使用随机算法错开（Raft 关键实现），避免无谓的死锁。
 * 选举者优先选举自己,将自己变成 candidate。
 * 选举的第一步就是把自己的 term 加一。
 * 然后像其他节点发送请求投票 RPC，请求参数参照论文，包括自身的 term，自身的 lastIndex，以及日志的 lastTerm。同时，请求投票 RPC 应该是并行请求的。
 * 等待投票结果应该有超时控制，如果超时了，就不等待了。
 * 最后，如果有超过半数的响应为 success，那么就需要立即变成 leader ，并发送心跳阻止其他选举。
 * 如果失败了，就需要重新选举。注意，这个期间，如果有其他节点发送心跳，也需要立刻变成 follower，否则，将死循环。
 */

@Getter
@Setter
public class NodeImpl<T> implements Node<T> {



    public RpcClient rpcClient = new DefaultRpcClient();

    volatile long currentTerm = 0;
    volatile String votedFor;

    public volatile int status = NodeStatus.FOLLOWER;
    public PeerSet peerSet;

    /** 日志条目集；每一个条目包含一个用户状态机执行的指令，和收到时的任期号 */
    LogModule logModule;

    Consensus consensus;


    /** 上一次心跳 */
    public volatile long preHeartBeat = 0;
    /** 心跳间隔 */
    public volatile long HeartBeatTick = 5*1000;

    /**  上一次心跳时间   */
    public volatile long preElection = 0;

    /** 选举间隔*/
    public volatile long ElectionTick = 7*1000;

    /** 同步索引*/
    ConcurrentHashMap<String,Long> nextIndex = new ConcurrentHashMap<>();
    ConcurrentHashMap<String,Long> matchIndex = new ConcurrentHashMap<>();



    public volatile boolean started;
    public static RpcServer RPCserver;
    HeartBeatTask heartBeatTask = new HeartBeatTask();
    ElectionTask electionTask = new ElectionTask();



    StateMachineImpl stateMachine;
    //最后应用于状态机的日志条目索引值。
    volatile long  lastApplied = 0;
    volatile long commitIndex;
    /**
     *  单例
     * @param
     */
   private NodeImpl(){}

    public static NodeImpl getInstance(){return NodeLazyHolder.INSTANCE;}

    private static class NodeLazyHolder{
       private static final NodeImpl INSTANCE = new NodeImpl();
    }

    @Override
    public void init()   {

       if(started)return;

       synchronized (this){
           if (started)return;

           RPCserver.start();
           consensus = new ConsensusImpl(this);
           //todo delegate

           RaftThreadPool.scheduleWithFixedDelay(heartBeatTask,500);
           RaftThreadPool.scheduleAtFixedRate(electionTask,6000,500);
           //todo 复制队列消费

           LogEntry logEntry = logModule.getLast();
           if(logEntry!=null){
               currentTerm = logEntry.getTerm();
           }
       }


    }

    public NodeConfig config;

    @Override
    public void setConfig(NodeConfig config) {
           this.config = config;
           stateMachine  = StateMachineImpl.getInstance();
           logModule = LogModuleImpl.getInstance();

            peerSet = PeerSet.getInstance();
            for (String s : config.getPeerAddrs()){
                peerSet.addPeer(s);
                if(s.equals("localhost:"+config.getSelfPort())){
                    peerSet.setSelf(s);
                }

            }

            RPCserver=new DefaultRpcServer(config.selfPort,this);

    }

    @Override
    public RvoteResult handleReqVote(RvoteParam param) {
        return consensus.requestVote(param);
    }

    @Override
    public AentryResult handleAppendEntries(AentryParam param) {
        return consensus.appendEntries(param);
    }


    /**
     * 每个节点都可能会接收到客户端的请求，但只有 leader 能处理，所以如果自身不是 leader，则需要转发给 leader。
     * 然后将用户的 KV 数据封装成日志结构，包括 term，index，command，预提交到本地。
     * 并行的向其他节点发送数据，也就是日志复制。
     * 如果在指定的时间内，过半节点返回成功，那么就提交这条日志。
     * 最后，更新自己的 commitIndex，lastApplied 等信息。
     * */



    @Override
    public ClientKVAck handlerClientRequest(ClientKVReq request) {

        if(status!=NodeStatus.LEADER){
            redirect(request);
        }

        if(request.getType() == ClientKVReq.GET){
            LogEntry logEntry = stateMachine.get(request.getKey());
            if (logEntry!=null)return new ClientKVAck(logEntry.getCommand() ) ;

            return new ClientKVAck(null);
        }

        LogEntry logEntry = LogEntry.newBuilder().term(currentTerm).Command(new Command(request.getKey(),request.getValue())).builder();

        logModule.write(logEntry);

        List<Future<Boolean>>futuresList = new ArrayList<>();
        //向各个节点添加日志
        for (String add : peerSet.getPeersWithOutSelf()){
            futuresList.add(replication(add,logEntry));
        }


        CountDownLatch latch = new CountDownLatch(futuresList.size());
        List<Boolean>resList = new CopyOnWriteArrayList<>();

        getRpcAppendResult(futuresList,latch,resList);
        AtomicInteger atomicInteger = new AtomicInteger(0);
        try {
            //阻塞当期线程 直到latch锁的计数器为0  或 超时
            latch.await(4000,MILLISECONDS);
        }catch (InterruptedException e){
            e.printStackTrace();
        }

        //
        resList.stream().forEach(i->{
            if(i == true)atomicInteger.incrementAndGet();
        });


        //??????
        List<Long> matchIndexList = new ArrayList<>(matchIndex.values());
        int median = 0;
        if(matchIndexList.size()>=2){
            Collections.sort(matchIndexList);
            median = matchIndexList.size()/2;
        }
        Long N = matchIndexList.get(median);
        if(N>commitIndex){
            LogEntry entry = logModule.read(N);
            if( entry!=null && entry.getTerm()==currentTerm ){
                commitIndex = entry.getIndex();
            }
        }





        //更新提交 正式提交到状态机
        if(atomicInteger.get()>futuresList.size()/2){
                commitIndex = logEntry.getIndex();
                getStateMachine().apply(logEntry);
                return ClientKVAck.ok();
        }else {
            logModule.removeOnStarIndex(logEntry.getIndex());
        }

        return null;
    }


    private void getRpcAppendResult( List<Future<Boolean>>futuresList,CountDownLatch latch, List<Boolean> resList){

        for(Future<Boolean> future : futuresList){
            RaftThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        resList.add(future.get(3000,MILLISECONDS));
                    }catch ( InterruptedException | ExecutionException | TimeoutException e){
                        e.printStackTrace();
                    }finally {
                        latch.countDown();
                    }
                }
            });
        }

    }


    private Future<Boolean> replication(String add , LogEntry logEntry){



        return RaftThreadPool.submit(new Callable() {
            @Override
            public Boolean call() throws Exception {

                long start = System.currentTimeMillis(),end = start;

                while (end-start<20*1000L){
                    AentryParam param = new AentryParam();
                    param.setTerm(currentTerm);
                    param.setEntries(new LogEntry[]{logEntry});
                    param.setServiceId(add);
                    param.setLeaderId(peerSet.getLeader());
                    param.setLeaderCommitIndex(commitIndex);
                    Long nextIndexs = nextIndex.get(add);

                    //打包 从节点 落下的数组
                    ArrayList<LogEntry> logEntryArrayList = new ArrayList<>();
                    logEntryArrayList.add(logEntry);


                    if(nextIndexs<logEntry.getIndex()){

                        for(long i = nextIndexs;i<logEntry.getIndex();i++){
                            LogEntry read = logModule.read(i);
                            if(read!=null){
                                logEntryArrayList.add(read);
                            }
                        }

                    }

                    //找到上一条数据 便于对比
                    LogEntry preLog = getPreLog(logEntryArrayList.get(0));
                    param.setPreLogTerm(preLog.getTerm());
                    param.setPrevLogIndex(preLog.getIndex());

                    param.setEntries(logEntryArrayList.toArray(new LogEntry[0]));


                    Request request = new Request(Request.A_ENTRIES,param,add);

                    Response response = getRpcClient().send(request);
                    if(response == null)return false;

                    AentryResult result = (AentryResult) response.getResult();

                    //返回成功 就更新从节点的当期最高索引  和 下一个索引节点
                    if(result!=null&& result.isSuccess()){
                        nextIndex.put(add,logEntry.getIndex()+1);
                        matchIndex.put(add,logEntry.getIndex());
                        return true;
                    }else if(result !=null) {
                         //对方任期比自己大 成为Follower
                            if(result.getTerm()>currentTerm){
                                currentTerm = result.getTerm();
                                status = NodeStatus.FOLLOWER;
                            }else {
                                //递减 重试
                                if(nextIndexs == 0){
                                    nextIndexs = 1L;
                                }
                                nextIndex.put(add, nextIndexs - 1);
                            }
                    }

                    end = System.currentTimeMillis();

                }


                return false;
            }
        });
    }


    public LogEntry getPreLog(LogEntry logEntry){

        LogEntry entry = logModule.read(logEntry.getIndex() - 1);
        if(entry == null){
            entry = LogEntry.newBuilder().index(0).term(0).Command(null).builder();
        }
        return  entry;

    }

    @Override
    public ClientKVAck redirect(ClientKVReq request) {
        Request<ClientKVReq> req = new Request<ClientKVReq>(Request.CLIENT_REQ,request,peerSet.getLeader());
        Response send = rpcClient.send(req);

        return (ClientKVAck) send.getResult();
    }


    @Override
    public void destroy() throws Throwable {
        RPCserver.stop();
    }



    //心跳检测
    class HeartBeatTask implements Runnable{

        @Override
        public void run() {

            if(status!=NodeStatus.LEADER)return;

            //还没到时间节点就不发送心跳了
            if(System.currentTimeMillis() - preHeartBeat < HeartBeatTick ){
                return;
            }

            preHeartBeat = System.currentTimeMillis();

            //向从节点发送心跳消息
            for(String add : peerSet.getPeersWithOutSelf()){

                AentryParam param = AentryParam.newBuilder()
                        .entries(null)
                        .leaderId(peerSet.getSelf())
                        .serverId(add)
                        .term(currentTerm)
                        .build();


                Request<AentryParam> request = new Request<>(Request.A_ENTRIES,param,add);


                //发送请求 异步进行
                RaftThreadPool.execute(()->{
                     //发送请求 异步进行
                    Response response = getRpcClient().send(request);
                    //接收返回的请求
                    AentryResult result = (AentryResult) response.getResult() ;

                    //如果任期比自己大 让主
                    if(result.getTerm()>currentTerm){
                        status = NodeStatus.FOLLOWER;
                        currentTerm = result.getTerm();
                        votedFor = "";
                    }

                },false);

            }



        }
    }


    //如果主机宕机  -》  选举
    class ElectionTask implements Runnable {

        @Override
        public void run() {

            if(status==NodeStatus.LEADER){
                return;
            }

            //检测  如果长时间没收到心跳 更新选举时间 就重新进行选举

            //通过随机算法  错开各节点选举时间 避免死锁
            ElectionTick = ElectionTick + ThreadLocalRandom.current().nextInt(50);
             if(System.currentTimeMillis() - preElection < ElectionTick) {
                 return;
             }

             preElection = System.currentTimeMillis();

             //重新选举    状态更为候选者  当前任期+1  投票给自己。
             status = NodeStatus.CANDIDATE;
             currentTerm +=1;
             votedFor = peerSet.getSelf();

            ArrayList<Future>futures = new ArrayList<>();
            for (String add : peerSet.getPeersWithOutSelf()){
                //发送投票请求  并接收响应
              futures.add(RaftThreadPool.submit( ()->{
                  //构建投票请求
                  long lastTerm = 0;
                  LogEntry last = logModule.getLast();
                  if(last!=null){
                      lastTerm = last.getTerm();
                  }

                  RvoteParam param = RvoteParam.newBuilder().candidateId(peerSet.getSelf()).serviceId(add).term(currentTerm)
                          .lastLogIndex(logModule.getLastIndex()).lastLogTerm(lastTerm).build();

                  Request<RvoteParam> request = new Request<>(Request.R_VOTE,param,add);

                  try {
                      Response<RvoteResult> response = getRpcClient().send(request);
                      return response;
                  }catch (RaftRemotingException e){
                      e.printStackTrace();
                      return null;
                  }

              } ) )  ;

            }

            AtomicInteger atomicInteger = new AtomicInteger(0);
            CountDownLatch latch = new CountDownLatch(futures.size());

            //筛选选票
            for(Future futu : futures){
                RaftThreadPool.submit(()->{

                    try {

                        Response<RvoteResult> result = (Response<RvoteResult>) futu.get(3000,MILLISECONDS);
                        if (result == null){
                            return null;
                        }

                        if(result.getResult().isVote()){
                            atomicInteger.incrementAndGet();
                        }else {
                            if(result.getResult().getTerm()>currentTerm){
                                currentTerm = result.getResult().getTerm();
                            }
                        }


                    }catch (Exception e){
                        e.printStackTrace();
                    }finally {
                        latch.countDown();
                    }

                    return 0;
                });
            }
            //等待筛选结束
            try {
                latch.await(5000,MILLISECONDS);
            }catch (Exception e){
                e.printStackTrace();
            }

            //如果状态已经变为跟随者 就停止选举
            if (status == NodeStatus.FOLLOWER){
                return;
            }

            if(atomicInteger.get()>=peerSet.getList().size()/2){
                status = NodeStatus.LEADER;
                peerSet.setLeader(peerSet.getSelf());
                votedFor = "";
                //todo 执行成为领导人做的事情
                toBeLeader();

            }else {
               votedFor = "";
            }

        }
    }


     private void  toBeLeader(){
            //重置索引
          nextIndex = new ConcurrentHashMap<>();
          matchIndex = new ConcurrentHashMap<>();

          for(String  add : peerSet.getPeersWithOutSelf()){
              nextIndex.put(add,logModule.getLastIndex()+1);
              matchIndex.put(add,0L);
          }
    }


}
